import pymysql
import numpy as np
import pandas as pd
from sqlalchemy import create_engine
import time, datetime, os

if __name__ == '__main__':
    # 本地数据库连接信息（已确认正确）
    # engine = create_engine('mysql+pymysql://root:123456@localhost:3306/trace')
    engine = create_engine('mysql+pymysql://5gzhyy:B6.5gzhyy312@132.91.175.98:8067/g41_wyzx_5gzhyydb')
    print(time.strftime(r"%Y-%m-%d %H:%M:%S", time.localtime()))

    from datetime import datetime, timedelta

    # 获取当前时间的前两天（上周六，对应原手动执行的times1_str）和前九天（上上周六）
    # times1_str = (datetime.now() - timedelta(days=2)).strftime("%Y%m%d")  # 上周六（目标统计日期）
    # times2_str = (datetime.now() - timedelta(days=9)).strftime("%Y%m%d")  # 上上周六

    # 手动执行时可注释上方，启用下方固定日期（示例）
    times1_str = '20250920'   # 手动指定上周六日期
    times2_str = '20250913'   # 手动指定上上周六日期


    # 日期相关计算（保持原逻辑不变）
    times1 = datetime.strptime(times1_str, "%Y%m%d")
    this_month_str = times1.strftime("%Y%m")
    first_day_of_month = times1.replace(day=1)
    day_of_month = times1.day
    week_number = (day_of_month - 1) // 7 + 1
    last_month_day_one = (first_day_of_month - timedelta(days=1)).replace(day=1)
    last_month_year_month = last_month_day_one.strftime("%Y%m")

    # 打印日期信息（验证用）
    print(f"这个月: {this_month_str}")
    print(f"上个月: {last_month_year_month}")
    print(f"上周六日期 {times1_str} 在本月的相对位置是第 {week_number} 周")
    print("前两天日期（根据上周六计算）:", (datetime.strptime(times1_str, "%Y%m%d") - timedelta(days=2)).strftime("%Y%m%d"))
    print("前九天日期（根据上周六计算）:", (datetime.strptime(times1_str, "%Y%m%d") - timedelta(days=9)).strftime("%Y%m%d"))


    # CSV文件路径（本地Windows路径，已修正转义）
    # file_path_1 = 'C:\\work\\data\\tongji'  # 注意：Windows路径用双反斜杠，这里简化为1个反斜杠也可（Python会自动处理）
    file_path_1 = '/oss_luoshen/ywhx/SVIP_gz_table/tongji/'
    file_name_1 = 'svip8_tongji_'
    file_name = file_name_1 + times1_str + '_zf.csv'
    print(f"生成的CSV文件名: {file_name}")

    # 核心SQL：新增this_data_poor字段（统计上周六4个字段中≥2个大于2的用户数）
    sql_query = f"""
          select  {times1_str} as stat_date, {this_month_str} as month,
          {week_number} AS week_of_month,
          (select count(*) from svip8_yw_feedback where END_DATE = {times1_str}) as this_week_svipuser,
          (select count(distinct a.msisdn) from (select msisdn from svip8_yw_feedback where end_date = {times1_str}) a INNER JOIN (select msisdn from svip8_yw_fromkefu where end_date = {times2_str}) b ON a.msisdn = b.msisdn) as this_trace_svipuser,
          (select count(*)  from (select msisdn,
          CASE 
              WHEN SUM(abnormal_events) = 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '2'
              WHEN SUM(abnormal_events) > 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '1'
          END AS priority 
          from svip8_feed_to_wy where end_date = {times1_str} group by msisdn)a where a.priority=1) as priority1,
          (select count(*) from (select msisdn,
          CASE 
              WHEN SUM(abnormal_events) = 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '2'
              WHEN SUM(abnormal_events) > 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '1'
          END AS priority 
          from svip8_feed_to_wy where end_date = {times1_str} group by msisdn)a where a.priority=2) as priority2,
          (select count(*) from (select msisdn,
          CASE 
              WHEN SUM(abnormal_events) = 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '2'
              WHEN SUM(abnormal_events) > 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '1'
          END AS priority 
          from svip8_feed_to_wy where end_date = {times1_str} group by msisdn)a) as priorityall,	
          (select count(*) from (select msisdn,
          CASE 
              WHEN SUM(abnormal_events) = 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '2'
              WHEN SUM(abnormal_events) > 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '1'
          END AS priority 
          from svip8_feed_to_wy where end_date = {times1_str} group by msisdn)a where a.priority=1) as this_poor_svipuser,
          (select count(distinct msisdn) from (select msisdn from svip8_yw_feedback where end_date = {times2_str} and abnormal_events >= 2 and abnormal_times / NULLIF(call_times, 0) >= 0.01 
              and msisdn in (select msisdn from svip8_yw_feedback where end_date = {times1_str}))a) as last_poo_svipuser_inthis,
          (select count(distinct msisdn) from (select msisdn from svip8_yw_feedback where end_date = {times2_str} and abnormal_events >= 2 and abnormal_times / NULLIF(call_times, 0) >= 0.01 
              and msisdn in (select msisdn from svip8_yw_feedback where end_date = {times1_str} and abnormal_events=0))a)  as last_poo_svipuser_better_inthis,
          (select count(*)  from svip8_feed_to_wy where end_date = {times1_str}) as this_poor_wy,
          (select count(distinct a.msisdn)  from svip8_feed_to_wy a where a.priority_week LIKE '%%/%%' and month = {last_month_year_month} and a.msisdn IN (select b.msisdn from svip8_yw_feedback b where end_date = {times1_str} GROUP BY b.msisdn)) as last_poor_wy_inthis,
          (select count(distinct a.msisdn) from (select msisdn from svip8_feed_to_wy where priority_week LIKE '%%/%%' and month = {last_month_year_month}) a join (select msisdn from svip8_yw_feedback where end_date = {times1_str} and ABNORMAL_EVENTS = 0) b on a.msisdn = b.msisdn) as last_poor_wy_better_inthis
           ;"""  # 已移除最后三个统计项
    print(sql_query)
    # 执行SQL查询并写入数据库/CSV
    df = pd.read_sql_query(sql_query, engine)
    df.to_sql('svip8_tongji', engine, if_exists='append', index=False, chunksize=100)

    if not df.empty:
        full_file_path = os.path.join(file_path_1, file_name)
        df.to_csv(full_file_path, index=False, encoding='utf-8')  # 导出CSV
        os.chmod(full_file_path, 0o777)  # 赋予文件读写权限（Windows下可注释，不影响）
        print(f"数据已成功追加到数据库，且CSV文件已保存至：{full_file_path}")
    else:
        print(f"上周六 {times1_str} 无数据，未生成CSV和写入数据库")